package com.stu.rocketmq.syncmessage;

import com.alibaba.fastjson.JSONObject;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.UUID;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

public class RocketMQProducer {

    private static Logger logger = LoggerFactory.getLogger(RocketMQProducer.class);

    public static void main(String[] args) throws MQClientException {
        logger.info("开始执行定时任务：{}", LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss:SSS")));
        ThreadFactory threadFactory = new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r,"producer-thread");
            }
        };
        final ScheduledExecutorService scheduledExecutorService = new  ScheduledThreadPoolExecutor(3,threadFactory);
        scheduledExecutorService.scheduleWithFixedDelay(new Task(),0,3, TimeUnit.SECONDS);
    }

   static class Task implements Runnable{

        private static DefaultMQProducer producer;

        static {
            producer =  new DefaultMQProducer("sync-producer-group");
            producer.setNamesrvAddr("172.168.0.101:9876;172.168.0.103:9876");
            producer.setInstanceName("sync-producer-instance");
            producer.setSendMsgTimeout(2000);
            try {
                producer.start();
            } catch (MQClientException e) {
                throw new RuntimeException(e);
            }
        }

        public Task(){
        }
       @Override
       public void run() {
           int number = 0;
           try {
               logger.info("{},开始发送消息",producer.getInstanceName());
               for (int i = 0;i<1;i++){
                   String messageKey = UUID.randomUUID().toString().replaceAll("-", "");
                   User user = new User(messageKey, "zhangsan"+i,10+i);
                   String userJson = JSONObject.toJSONString(user);
                   Message message = new Message("sync-message-topic","tag-a",messageKey,userJson.getBytes());
                   producer.send(message);
                   logger.info("{}-{},已发送的消息：{}",Thread.currentThread().getName(),Thread.currentThread().getId(),messageKey);
                   number++;
               }
           } catch (MQClientException e) {
               logger.error("rocketmq 客户端异常:{}",e.getMessage());
           } catch (RemotingException e) {
               logger.error("rocketmq 客户端连接异常：{}",e.getMessage());
           } catch (MQBrokerException e) {
               logger.error("broker 连接异常：{}",e.getMessage());
           } catch (InterruptedException e) {
               logger.error("线程中断异常：{}",e.getMessage());
           }finally {
               logger.info("已发送完成{}条消息",number);
           }
       }
   }

    static class User{
        private String userId;
        private String userName;
        private int userAge;

        public User() {
        }

        public User(String userId, String userName, int userAge) {
            this.userId = userId;
            this.userName = userName;
            this.userAge = userAge;
        }

        public String getUserId() {
            return userId;
        }

        public void setUserId(String userId) {
            this.userId = userId;
        }

        public String getUserName() {
            return userName;
        }

        public void setUserName(String userName) {
            this.userName = userName;
        }

        public int getUserAge() {
            return userAge;
        }

        public void setUserAge(int userAge) {
            this.userAge = userAge;
        }

        @Override
        public String toString() {
            return "User{" +
                    "userId='" + userId + '\'' +
                    ", userName='" + userName + '\'' +
                    ", userAge='" + userAge + '\'' +
                    '}';
        }
    }

}
